package com.homedo.async.mysql.support.mapper.select;

import com.homedo.async.common.dto.Pagination;
import com.homedo.async.mysql.support.dml.DMLEnum;
import com.homedo.async.mysql.support.mapper.util.CreateFlux4RecordResultUtil;
import com.homedo.async.mysql.support.meta.EntityMetaData;
import com.homedo.async.mysql.support.prepare.PrepareObject2Sql;
import com.homedo.async.mysql.support.prepare.PrepareObjectType;
import com.homedo.async.mysql.support.prepare.PrepareRecords2Sql;
import com.homedo.plainly.logger.Logger;
import com.homedo.plainly.logger.LoggerFactory;
import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.SqlConnection;
import org.reactivestreams.Publisher;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @author jgshun
 * @date 2020/3/19 1:16 PM
 */
public class SelectPaginationMapperImpl<T, ID> implements SelectPaginationMapperI<T, ID> {
    private static final Logger _LOGGER = LoggerFactory.getLogger(SelectPaginationMapperImpl.class);

    private Pool pool;

    private EntityMetaData<T> entityMetaData;

    public SelectPaginationMapperImpl(Pool pool, EntityMetaData<T> entityMetaData) {
        this.pool = pool;
        this.entityMetaData = entityMetaData;
    }

    @Override
    public Flux<T> findAll(Pagination<T> pagination) {
        PrepareObject2Sql prepareObject2Sql = new PrepareObject2Sql(null, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, false);
        prepareObject2Sql.setPagination(pagination);

        Object[] prepareSqlArgs = prepareObject2Sql.prepare();

        return this.createSelectFlux(prepareSqlArgs, null);
    }

    @Override
    public Flux<T> findByRecordSelective(T entity, Pagination<T> pagination) {
        return this.findByRecord(entity, false, pagination, null);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecord(entityStream, false, pagination, null);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities, Pagination<T> pagination) {
        return this.findByRecords(entities, false, pagination, null);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecords(entityStream, false, pagination, null);
    }

    @Override
    public Flux<T> findByRecord(T entity, Pagination<T> pagination) {
        return this.findByRecord(entity, true, pagination, null);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecord(entityStream, true, pagination, null);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities, Pagination<T> pagination) {
        return this.findByRecords(entities, true, pagination, null);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecords(entityStream, true, pagination, null);
    }


    //==================

    @Override
    public Flux<T> findAll(Pagination<T> pagination, SqlConnection connection) {
        PrepareObject2Sql prepareObject2Sql = new PrepareObject2Sql(null, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, false);
        prepareObject2Sql.setPagination(pagination);

        Object[] prepareSqlArgs = prepareObject2Sql.prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(T entity, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecord(entity, false, pagination, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecord(entityStream, false, pagination, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecords(entities, false, pagination, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecords(entityStream, false, pagination, connection);
    }

    @Override
    public Flux<T> findByRecord(T entity, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecord(entity, true, pagination, connection);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecord(entityStream, true, pagination, connection);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecords(entities, true, pagination, connection);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.findByRecords(entityStream, true, pagination, connection);
    }

    private Flux<T> createSelectFlux(Object[] prepareSqlArgs, SqlConnection connection) {
        return CreateFlux4RecordResultUtil.create(prepareSqlArgs, this.pool, this.entityMetaData.getCollector(), connection);
    }

    private Flux<T> findByRecord(T entity, boolean includeNullVals, Pagination<T> pagination, SqlConnection connection) {
        Assert.notNull(entity, "The given entity must not be null!");

        PrepareObject2Sql prepareObject2Sql = new PrepareObject2Sql(entity, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, includeNullVals);
        prepareObject2Sql.setPagination(pagination);

        Object[] prepareSqlArgs = prepareObject2Sql.prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    private Flux<T> findByRecord(Publisher<? extends T> entityStream, boolean includeNullVals, Pagination<T> pagination, SqlConnection connection) {
        Assert.notNull(entityStream, "The given Publisher of entity must not be null!");

        return Flux.create(sink -> {
            Mono.from(entityStream).subscribe(e -> {
                this.findByRecord(e, includeNullVals, pagination, connection).subscribe(res -> {
                    sink.next(res);
                }, error -> sink.error(error), new Runnable() {
                    @Override
                    public void run() {
                        sink.complete();
                    }
                });
            }, error -> sink.error(error));
        });
    }

    private Flux<T> findByRecords(Iterable<? extends T> entities, boolean includeNullVals, Pagination<T> pagination, SqlConnection connection) {
        Assert.notNull(entities, "The given Iterable of entity's must not be null!");
        PrepareRecords2Sql prepareRecords2Sql = new PrepareRecords2Sql(entityMetaData, DMLEnum.SELECT, entities, includeNullVals);
        prepareRecords2Sql.setPagination(pagination);

        Object[] prepareSqlArgs = prepareRecords2Sql.prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    private Flux<T> findByRecords(Publisher<? extends T> entityStream, boolean includeNullVals, Pagination<T> pagination, SqlConnection connection) {
        Assert.notNull(entityStream, "The given Publisher of entity's must not be null!");

        return Flux.from(entityStream).buffer().flatMap(e -> {
            return this.findByRecords(e, includeNullVals, pagination, connection);
        });
    }

}
